import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Properties;

public class ProducerDemo {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("metadata.broker.list", "master:9092,slave1:9092,slave2:9092,slave3:9092");
        //System.out.println("行不行");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        ProducerConfig config = new ProducerConfig(props);
        kafka.javaapi.producer.Producer<String, String> producer = new kafka.javaapi.producer.Producer<String, String>(config);
        Integer count = 0;
        try {
            String line1 = null;
            BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream("F:\\大数据阶段\\大数据项目\\项目3-中国移动充值运营分析平台实时监控平台\\day01\\flumeLoggerapp4.log.20170412"), "UTF-8"));
            while ((line1 = br.readLine()) != null) {
                producer.send(new KeyedMessage<String, String>("recharges171116_3", line1));
                //Thread.sleep(1000L);
                count += 1 ;
                System.out.println(count);
            }

            br.close();

        } catch (IOException e) {
            e.printStackTrace();
        }

    }
}